package com.binance.rabbitmq.recoverer;

import com.binance.rabbitmq.config.DefaultConfig;
import com.binance.rabbitmq.policy.RecovererPolicy;
import lombok.extern.log4j.Log4j2;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.retry.MessageRecoverer;
import org.springframework.amqp.support.converter.MessageConversionException;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.TimeZone;

/**
 * 业务中消息处理失败，耗尽了spring-retry ，消息走MessageRecoverer
 * 根据recovererPolicy 来判断重试次数，如果可以重试，则发送至延迟Exchange，反之发送至失败Exchange
 * Created by robin.wu on 2018/11/29.
 */
@Log4j2
public class DelayMessageRecoverer implements MessageRecoverer {
    @Autowired
    private AmqpTemplate errorTemplate;

    @Autowired
    private DefaultConfig defaultConfig;

    @Autowired(required = false)
    private List<MessageExceptionHandler> messageExceptionHandlers=new ArrayList<>();

    @Autowired
    private RecovererPolicy deadLetterRecovererPolicy;


    @Override
    public void recover(Message message, Throwable cause) {
        if(cause != null && cause.getCause() instanceof MessageConversionException) {
            log.warn("MessageConversion exception will not send to delay or fail exchange. This often occurred when message is not valid. Message : {}",message.toString());
            return;
        }
        Map<String, Object> headers = message.getMessageProperties().getHeaders();
        headers.put("x-exception-stacktrace", ExceptionUtils.getStackTrace(cause));
        headers.put("x-exception-message", ExceptionUtils.getMessage(cause));
        headers.put("x-exception-root-cause-message", ExceptionUtils.getRootCauseMessage(cause));
        headers.put("x-original-exchange", message.getMessageProperties().getReceivedExchange());
        headers.put("x-original-routingKey", message.getMessageProperties().getReceivedRoutingKey());
        headers.put("x-original-queue", message.getMessageProperties().getConsumerQueue());
        headers.put("x-recoverer-time", DateFormatUtils.format(new Date(),"yyyy-MM-dd HH:mm:ss", TimeZone.getTimeZone("UTC")));

        String delayExchangeName = defaultConfig.getDelayConfig().getDelayExchange().getName();
        String delayRoutingKey = defaultConfig.getDelayConfig().getRoutingKey();

        String failExchangeName = defaultConfig.getFailConfig().getFailExchange().getName();
        String failRoutingKey = defaultConfig.getFailConfig().getRoutingKey();

        if(headers.containsKey("correlation-id")) {
            message.getMessageProperties().setCorrelationIdString((String) headers.get("correlation-id"));
        }

        headers.putAll(loadAdditionalHeaders(message, cause));

        for (MessageExceptionHandler messageExceptionHandler : messageExceptionHandlers) {
            try {
                messageExceptionHandler.handle(message, cause);
            } catch (Exception e) {
                log.error("Exception occurred while processing '{}' message exception handler.", messageExceptionHandler, e);
            }
        }

        //判断是否可以再重试
        if(deadLetterRecovererPolicy.recoverable(message)) {
            log.warn("Republishing failed message to delay exchange '{}', routing key '{}', message {{}} ",
                    delayExchangeName, delayRoutingKey, message);

            this.errorTemplate.send(delayExchangeName, delayRoutingKey, message);

        } else {
            log.warn("Republishing failed message to fail exchange '{}', routing key '{}', message {{}}",
                    delayExchangeName, delayRoutingKey, message);
            this.errorTemplate.send(failExchangeName,failRoutingKey,message);
        }
    }


    /**
     * 可以重写此方法来扩展header
     * @param message
     * @param cause
     * @return
     */
    protected Map<String, Object> loadAdditionalHeaders(Message message, Throwable cause) {
        log.info("No additional headers added for message {}, cause {}", message, cause == null ? null : cause.getMessage());
        return new HashMap<>();
    }
}
